package com.xxl.job.executor.sample.frameless.jobhandler;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.xuxueli.executor.carte.model.ExecutionResult;
import com.xuxueli.executor.carte.model.FileLoggingEventListener;
import com.xuxueli.executor.carte.model.TransExecutionConfiguration;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import okhttp3.*;
import org.pentaho.di.core.logging.KettleLogStore;
import org.pentaho.di.core.logging.LogLevel;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;

import java.io.IOException;
import java.net.URLDecoder;
import java.net.URLEncoder;
import java.util.HashMap;
import java.util.Map;

import static com.xxl.job.core.biz.model.ReturnT.FAIL;
import static com.xxl.job.core.biz.model.ReturnT.SUCCESS;

/**
 * 调用kettle，参数为脚本地址，和脚本参数
 * 先去脚本地址取脚本的内容，然后执行脚本
 * 脚本可以存在ftp服务器上，直接从上面取文件
 *
 * @author xuxueli 2015-12-19 19:43:36
 */
public class CarteJobHandler{
    private static ObjectMapper objectMapper = new ObjectMapper();
    private static Map<Long,Trans> transMap = new HashMap<>();

    @XxlJob(value="carteJobHandler",destroy="destroy")
    public ReturnT<String> carteJobHandler() throws Exception {
        String param = XxlJobHelper.getJobParam();
        XxlJobHelper.log("start execution of kettle script.");
        //1 param是个json，里面有任务对象id（taskObjId），有执行脚本地址(filename)，还有其他脚本变量(vars)
        TransExecutionConfiguration transExecutionConfiguration = objectMapper.readValue(param, TransExecutionConfiguration.class);
        //2 调用kettle脚本
        TransMeta transMeta = new TransMeta(URLDecoder.decode(transExecutionConfiguration.getFileUrl()));
        Trans trans = new Trans(transMeta);
        //2.1 怎么把执行日志输出到xxljob的执行日志里面
        String logFileName = XxlJobHelper.getJobLogFileName();
        FileLoggingEventListener loggingEventListener = new FileLoggingEventListener(trans.getLogChannelId(),logFileName,true);
        KettleLogStore.getAppender().addLoggingEventListener( loggingEventListener );
        //2.2 准备转换相关配置
        trans.setReplayDate( transExecutionConfiguration.getReplayDate() );
        trans.setSafeModeEnabled( transExecutionConfiguration.isSafeModeEnabled() );
        trans.setGatheringMetrics( transExecutionConfiguration.isGatheringMetrics() );
        for(Map.Entry<String,String> entity: transExecutionConfiguration.getParams().entrySet()){
            trans.setParameterValue(entity.getKey(),entity.getValue());
        }
        trans.setPreviousResult( transExecutionConfiguration.getPreviousResult() );
        trans.setLogLevel(null == transExecutionConfiguration.getLogLevel() ? LogLevel.BASIC:transExecutionConfiguration.getLogLevel());
        //2.3 开始执行
        final long jobId = XxlJobHelper.getJobId();
        try{
            transMap.put(jobId, trans);
            trans.prepareExecution(null);
            trans.startThreads();
            trans.waitUntilFinished();
        }finally {
            //2.4 执行完毕不再往日志文件中写日志
            KettleLogStore.getAppender().removeLoggingEventListener( loggingEventListener );
            transMap.remove(jobId);
        }
        final ReturnT<String> result = trans.getErrors() != 0 ? FAIL : SUCCESS;
        /**
         * 3. 不管成功还是失败，回调应用接口，通知执行情况;
         *   最好通知是写到kafka里面，这样应用可以关注kafka的通知，过滤jobId的通知然后修改数据库
         *   http的通知没有缓存，如果失败就丢失了
         */
        callback(transExecutionConfiguration,result.getCode());

        XxlJobHelper.log("complete execution of kettle script.");
        return result;
    }

    public void destroy(){
        final long jobId = XxlJobHelper.getJobId();
        Trans trans = transMap.get(jobId);
        if(null != trans) {
            trans.stopAll();
        }
    }


    private void callback(TransExecutionConfiguration transExecutionConfiguration, int code) throws JsonProcessingException {
        String url = transExecutionConfiguration.getCallbackUrl();
        if(null == url) return ;

        OkHttpClient okHttpClient = new OkHttpClient();
        MediaType mediaType = MediaType.parse("application/json; charset=utf-8");
        //把taskObjId和执行是否成功返回回去
        ExecutionResult result = new ExecutionResult();
        result.setTaskObjectId(transExecutionConfiguration.getTaskObjId());
        result.setCode(code);
        String requestBody = objectMapper.writeValueAsString(result);
        //1. 返回对象是执行结果
        RequestBody body = RequestBody.create(mediaType, requestBody);
        final Request request = new Request.Builder()
                .url(url)
                .post(body)
                .build();
        Call call = okHttpClient.newCall(request);
        call.enqueue(new Callback() {
            @Override
            public void onFailure(Call call, IOException e) {
                XxlJobHelper.log(e);
            }

            @Override
            public void onResponse(Call call, Response response) throws IOException {
                XxlJobHelper.log( "onResponse: " + response.body().string());
            }
        });
    }

}
